package com.atguigu.flink0922.chapter07;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/5 11:19
 */
public class Flink05_Window_Aggregate {
    public static void main(String[] args) throws Exception {
        final Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        
        env.setParallelism(2);
    
        env
            .socketTextStream("hadoop162", 9999)
            .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
                @Override
                public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                    for (String word : value.split(" ")) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                }
            })
            .keyBy(t -> t.f0)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .aggregate(new AggregateFunction<Tuple2<String, Long>, Long, Long>() {
                // 给累加器初始化值
                @Override
                public Long createAccumulator() {
                    System.out.println("createAccumulator...");
                    return 0L;
                }
    
                // 对累加器进行累加
                @Override
                public Long add(Tuple2<String, Long> value, Long accumulator) {
                    System.out.println("add...");
                    return accumulator + 1;
                }
    
                // 返回最终的结果, 当窗口关闭的时候, 调用这个方法, 返回结果
                @Override
                public Long getResult(Long accumulator) {
                    System.out.println("getResult...");
                    return accumulator;
                }
    
                // 合并: 只有在session窗口才会调用
                @Override
                public Long merge(Long a, Long b) {
                    System.out.println("merge...");
                    return a + b;
                }
            }, new WindowFunction<Long, String, String, TimeWindow>() {
                @Override
                public void apply(String key,
                                  TimeWindow window,
                                  Iterable<Long> input,
                                  Collector<String> out) throws Exception {
        
                }
            })
            .print();
            
        
        env.execute();
        
    }
}
